package com.shujia.tf

import org.apache.flink.streaming.api.scala._

object Demo2Agg {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val student: DataStream[String] = env.readTextFile("data/students.txt")

    val students: DataStream[Student] = student.map(line =>{
      val split: Array[String] = line.split(",")
      val id: String = split(0)
      val name: String = split(1)
      val age: Int = split(2).toInt
      val gender: String = split(3)
      val clazz: String = split(4)
      Student(id,name,age,gender,clazz)
    })

    students
      .keyBy(_.clazz)
      .sum(2) //指定下标聚合，或者指定字段sum("age)
        //.print()

    /**
      *KeyedStream.sum(0)
      * KeyedStream.sum("key")
      * KeyedStream.min(0)
      * KeyedStream.min("key")
      * KeyedStream.max(0)
      * KeyedStream.max("key")
      * KeyedStream.minBy(0)
      * KeyedStream.minBy("key")
      * KeyedStream.maxBy(0)
      * KeyedStream.maxBy("key")
      *
      *
      * max 和 maxBy 之间的区别在于 max 返回流中的最大值，但 maxBy 返回具有最大值的键， min 和 minBy 同理。
      *
      */

    students
      .keyBy(_.clazz)
        .minBy("age")
        .print()



    env.execute()

  }

  case class Student(id: String, name: String, age: Int, gender: String, clazz: String)

}
